# -*- coding: utf-8 -*-

''' transfer模块的测试类
    
    测试原则：逐步编织和完善测试网, 拒绝一步到位，也拒绝同样错误再次漏过.
        验证主功能走通并正确
        对于边界情况和异常流程，在发生错误时定位捕获. 初步时可只针对重要情况做事先测试，也可不做。
'''

from unittest import TestCase
from wolfox.foxit.lib.mock import Mock,patch,sentinel
from wolfox.foxit.dune.test_common import prepare,clean
from wolfox.foxit.dune.store_test import prepare_xit

from wolfox.foxit.base.common import XInfo
from wolfox.foxit.source.transfer import *
from wolfox.foxit.source.fxj import *
from wolfox.foxit.source.reader import *
from wolfox.foxit.dune.store import NormalStore
from wolfox.foxit.dune.models import Catalog,StockCode,XDayQuote,SDayQuote

import os.path
cur_dir = os.path.dirname(__file__)
test_data_dir = os.path.join(cur_dir,'test_data')

class TransferTest(TestCase):
    def setUp(self):
        prepare()

    def tearDown(self):
        clean()

    def test_init_catalog(self): #包装方法,只测试通透性
        from django.db import connection        
        transfer = Transfer(NormalStore(),connection,0,99999999)        
        test_blk_dir = os.path.join(test_data_dir,'blk')
        test_code_file = os.path.join(test_data_dir,'SZ.SNT')
        creader = CatalogReader()
        store = NormalStore()
        new_catalogs = transfer.init_catalog(CodeReader(),[test_code_file],creader,[test_blk_dir])
        self.assertTrue(True)

    def test_sync(self):    #包装方法,只测试通透性和xbegin,xend. 测试self.transfer.set_quote_params,不设时间段
        from django.db import connection        
        transfer = Transfer(NormalStore(),connection,20070101,99999999)        
        test_c_files = (os.path.join(test_data_dir,'SZ.SNT'),os.path.join(test_data_dir,'SH.SNT'))
        test_q_files = (os.path.join(test_data_dir,'quote_test.dad'),os.path.join(test_data_dir,'quote_test2.dad'))
        test_h_files_range = [[os.path.join(test_data_dir,'m5_test.dad'),0,22000000]]
        test_x_files = (os.path.join(test_data_dir,'xtest.pwr'),)
        test_r_files = (os.path.join(test_data_dir,'report_test.fin'),)
        transfer.set_code_params(CodeReader(),test_c_files)
        transfer.set_quote_params(TradeExtractor(),test_q_files)
        transfer.set_xinfo_params(XExtractor(),test_x_files)
        transfer.set_report_params(ReportExtractor(),test_r_files)
        transfer.set_hour_quote_params2(M5to60Extractor(),test_h_files_range)
        cursor = transfer.conn.cursor()
        cursor.execute('delete from dune_xdayquote')
        cursor.execute('delete from dune_sdayquote')
        xbegin,xend = transfer.sync()
        self.assertTrue(xbegin <= xend)

    def test_sync60(self):    #包装方法,只测试通透性和xbegin,xend. 测试self.transfer.set_quote_params,不设时间段
        from django.db import connection        
        transfer = Transfer(NormalStore(),connection,20070101,99999999)        
        test_h_files_range = [[os.path.join(test_data_dir,'m5_test.dad'),0,22000000]]
        transfer.set_hour_quote_params2(M5to60Extractor(),test_h_files_range)
        cursor = transfer.conn.cursor()
        cursor.execute('delete from dune_xhourquote')
        cursor.execute('delete from dune_shourquote')
        xbegin,xend = transfer.sync60()
        self.assertTrue(xbegin <= xend)


    def test_sync2(self):    #包装方法,只测试通透性和xbegin,xend,测试self.transfer.set_quote_params2,每个行情文件都单设时间段
        from django.db import connection        
        transfer = Transfer(NormalStore(),connection,0,99999999)        
        test_c_files = (os.path.join(test_data_dir,'SZ.SNT'),os.path.join(test_data_dir,'SH.SNT'))
        test_q_files = (os.path.join(test_data_dir,'quote_test.dad'),os.path.join(test_data_dir,'quote_test2.dad'))
        test_x_files = (os.path.join(test_data_dir,'xtest.pwr'),)
        test_r_files = (os.path.join(test_data_dir,'report_test.fin'),)
        test_q_file_ranges = (test_q_files[0],20070101,20070801),(test_q_files[1],20080701,20080801)
        transfer.set_code_params(CodeReader(),test_c_files)
        transfer.set_quote_params2(TradeExtractor(),test_q_file_ranges)
        transfer.set_xinfo_params(XExtractor(),test_x_files)
        transfer.set_report_params(ReportExtractor(),test_r_files)
        cursor = transfer.conn.cursor()
        cursor.execute('delete from dune_xdayquote')
        cursor.execute('delete from dune_sdayquote')
        xbegin,xend = transfer.sync()
        self.assertTrue(xbegin <= xend)
    

class ModuleTest(TestCase):

    def test_create_date_filter(self):
        from wolfox.foxit.source.transfer import _create_date_filter
        filter1 = _create_date_filter(0,99999999)
        filter2 = _create_date_filter(20010101,20060101)
        self.assertEquals(NullFilter,type(filter1))
        self.assertEquals(DateFilter,type(filter2))

    def test_create_filter(self):
        from wolfox.foxit.source.transfer import _create_filter        
        filter1 = _create_filter(0,99999999,['SH01','SH02'])
        filter2 = _create_filter(20010101,20080101,['SH01','SH02'])
        self.assertEquals(CodeFilter,type(filter1))
        self.assertEquals(CodeFilter,type(filter2))

    def test_read_quotes(self):
        from wolfox.foxit.source.transfer import _read_quotes
        extractor = TradeExtractor()
        reader = Reader(extractor,NullFilter())
        quotes,begin,end = _read_quotes(os.path.join(test_data_dir,'quote_test.dad'),reader)
        self.assertEquals(71,len(quotes))
        self.assertEquals(20070702,begin)
        self.assertEquals(20070802,end)

    def test_read_records(self):    #以xinfo为例验证records
        from wolfox.foxit.source.transfer import _read_records
        extractor = XExtractor()
        reader = Reader(extractor,NullFilter())
        records,begin,end = _read_records(os.path.join(test_data_dir,'xtest.pwr'),reader)
        self.assertEquals(87,len(records))
        self.assertEquals(19981203,begin)
        self.assertEquals(20061114,end)

    def test_read_records(self):    #以xinfo为例验证records在数据为空时的情况
        from wolfox.foxit.source.transfer import _read_records,_create_date_filter
        extractor = XExtractor()
        reader = Reader(extractor,_create_date_filter(0,0))
        records,begin,end = _read_records(os.path.join(test_data_dir,'xtest.pwr'),reader)
        self.assertEquals(0,len(records))
        self.assertEquals(0,begin)
        self.assertEquals(0,end)

    def test_read_xinfos(self):    #通路测试
        from wolfox.foxit.source.transfer import _read_xinfos
        extractor = XExtractor()
        reader = Reader(extractor,NullFilter())
        records,begin,end = _read_xinfos(os.path.join(test_data_dir,'xtest.pwr'),reader)
        self.assertTrue(True)

    def test_read_reports(self):    #通路测试
        from wolfox.foxit.source.transfer import _read_reports
        extractor = ReportExtractor()
        reader = Reader(extractor,NullFilter())
        reports,begin,end = _read_reports(os.path.join(test_data_dir,'report_test.fin'),reader)
        self.assertTrue(True)

    @patch('wolfox.foxit.source.transfer','PowerInfo')
    def test_xit(self,pm):
        #prepare
        store = Mock()
        conn = Mock()
        store.get_latest_xquote_day.return_value=20090100
        filter_rev1 =Mock()
        filter_rev2 = Mock()
        record1,record2 = XInfo(),XInfo()
        pm.objects.filter.return_value = filter_rev1
        filter_rev1.filter.return_value = filter_rev2
        filter_rev2.order_by.return_value = (record1,record2)
        #action
        xit(store,conn,20010101)
        #assertion. 不过这个太侵入性了.渗入到实现的细节,未必是好. 一旦实现变化,这里也要变化.不若针对接口I/O得好
        self.assertEquals(True,pm.objects.filter.called)
        self.assertEquals(True,store.xit.called)
        self.assertEquals({'execute_day__range': (20010101, 20090100)},pm.objects.filter.call_args[1])
        self.assertEquals(conn,store.xit.call_args[0][0])
        self.assertEquals([record1,record2],store.xit.call_args[0][1])

    def test_xit_passthrough(self): #测试实际的通过性
        from django.db import connection
        prepare()
        prepare_xit(connection)
        xit(NormalStore(),connection,20010101)
        clean()

    def test_xit_hour_passthrough(self): #测试实际的通过性
        from django.db import connection
        prepare()
        prepare_xit(connection)
        xit_hour(NormalStore(),connection,20010101)
        clean()

    @patch('wolfox.foxit.source.transfer','PowerInfo')
    def test_xit2(self,pm):
        from wolfox.foxit.source.transfer import _xit2        
        #prepare
        store = Mock()
        conn = Mock()
        store.get_latest_xquote_day.return_value=20090100        
        filter1 = Mock()
        filter2 = Mock()
        filter3 = Mock()
        record1,record2 = XInfo(),XInfo()
        pm.objects.filter.return_value = filter1
        filter1.filter.return_value = filter2
        filter2.filter.return_value = filter3
        filter3.order_by.return_value = (record1,record2)
        #action
        _xit2(store,conn,['SH00000X','SH00000Y'],20010101,20090101)
        #assertion. 不过这个太侵入性了.渗入到实现的细节,未必是好. 一旦实现变化,这里也要变化.不若针对接口I/O得好
        self.assertEquals(True,pm.objects.filter.called)
        self.assertEquals(True,store.xit.called)
        self.assertEquals({'execute_day__range': (20010101, 20090100)},pm.objects.filter.call_args[1])
        self.assertEquals({'tstock__code__in': ['SH00000X', 'SH00000Y']},filter1.filter.call_args[1])
        self.assertEquals(conn,store.xit.call_args[0][0])
        self.assertEquals((record1,record2),store.xit.call_args[0][1])
    
    def test_xit2_passthrough(self): #测试实际的通过性
        from wolfox.foxit.source.transfer import _xit2        
        from django.db import connection
        prepare()
        prepare_xit(connection)
        _xit2(NormalStore(),connection,['SH00000X','SH00000Y'],20010101)
        clean()

    def test_sync_xinfos(self): #只检测实际通过性,间接测试sync_records
        extractor = XExtractor()
        store = NormalStore()
        length,begin,end = sync_xinfos(store,os.path.join(test_data_dir,'xtest.pwr'),extractor,0,99999999)
        self.assertTrue(length>0)

    def test_sync_reports(self): #只检测实际通过性,间接测试sync_records
        extractor = ReportExtractor()
        store = NormalStore()
        length,begin,end = sync_reports(store,os.path.join(test_data_dir,'report_test.fin'),extractor,0,99999999)
        self.assertTrue(length>0)

    def test_sync_codes(self):
        import os.path
        names = ['SZ.SNT','SH.SNT']
        test_data_files = map(lambda n: os.path.join(test_data_dir,n),names)
        creader = CodeReader()
        store = NormalStore()
        try:
            prepare()        
            new_codes = sync_codes(store,creader,test_data_files)
            new_codes2 = sync_codes(store,creader,test_data_files)
            wka = StockCode.objects.get(code='SZ000002')    #验证确实保存了，否则会抛出异常
        finally:
            clean()
        self.assertEquals(9,len(new_codes)) #SH000001已经在初始化时存在了
        self.assertEquals(0,len(new_codes2)) #第二次为没更新

    def test_init_catalog_local(self):  #测试_init_catalog，即本地版本
        from wolfox.foxit.source.transfer import _init_catalog
        test_blk_dir = os.path.join(test_data_dir,'blk')
        test_code_file = os.path.join(test_data_dir,'SZ.SNT')
        creader = CatalogReader()
        store = NormalStore()
        try:
            prepare()
            sync_codes(store,CodeReader(),[test_code_file])
            new_catalogs = _init_catalog(store,creader,[test_blk_dir]) 
            catalog_gd = Catalog.objects.get(name=u'广东')
            members = catalog_gd.stocks.values_list('code')
            self.assertEquals(2,len(members))            
            self.assertTrue(('SZ000429',) in members)
            self.assertTrue(('SZ000507',) in members)
        finally:
            clean()
        self.assertEquals(2,len(new_catalogs))
        self.assertTrue(u'广东' in new_catalogs)
        self.assertTrue(u'广西' in new_catalogs)

    def test_init_catalog(self):    #测试init_catalog，即对外版本
        test_data_dir = os.path.join(os.path.dirname(__file__),'test_data')
        test_blk_dir = os.path.join(test_data_dir,'blk')
        test_code_file = os.path.join(test_data_dir,'SZ.SNT')
        creader = CatalogReader()
        store = NormalStore()
        try:
            prepare()
            new_catalogs = init_catalog(store,CodeReader(),[test_code_file],creader,[test_blk_dir])
            catalog_gd = Catalog.objects.get(name=u'广东')
            members = catalog_gd.stocks.values_list('code')
            self.assertEquals(2,len(members))            
            self.assertTrue(('SZ000429',) in members)
            self.assertTrue(('SZ000507',) in members)
        finally:
            clean()
        self.assertEquals(2,len(new_catalogs))
        self.assertTrue(u'广东' in new_catalogs)
        self.assertTrue(u'广西' in new_catalogs)

    def test_sync_quotes(self): #通透性测试
        from django.db import connection
        store = NormalStore()
        test_quote_file = os.path.join(test_data_dir,'quote_test.dad')
        try:
            prepare()
            from sqlite3 import IntegrityError
            self.assertRaises(IntegrityError,sync_quotes,store,connection,test_quote_file,TradeExtractor(),0,99999999) #重复将导致异常
            cursor = connection.cursor()
            cursor.execute('delete from dune_xdayquote')
            cursor.execute('delete from dune_sdayquote')            
            xbegin,xend = sync_quotes(store,connection,test_quote_file,TradeExtractor(),0,99999999)
            xcount = XDayQuote.objects.filter(tdate__range=(xbegin,xend)).count()            
            scount = SDayQuote.objects.filter(tdate__range=(xbegin,xend)).count()
            self.assertRaises(IntegrityError,sync_quotes,store,connection,test_quote_file,TradeExtractor(),0,99999999) #重复将导致异常
        finally:
            clean()
        self.assertEquals(20070702,xbegin)
        self.assertEquals(20070802,xend)
        print 'xcount:%s,scount:%s' % (xcount,scount)
        self.assertEquals(23,xcount)
        self.assertEquals(48,scount)

    def test_sync_quotes2(self): #通透性测试
        from django.db import connection
        store = NormalStore()
        test_quote_file = os.path.join(test_data_dir,'quote_test.dad')
        try:
            prepare()
            xbegin,xend = sync_quotes2(store,connection,test_quote_file,TradeExtractor(),0,99999999)
            xcount = XDayQuote.objects.filter(tdate__range=(xbegin,xend)).count()            
            scount = SDayQuote.objects.filter(tdate__range=(xbegin,xend)).count()
            xbegin,xend = sync_quotes2(store,connection,test_quote_file,TradeExtractor(),0,99999999)    #再次倒入也不会出错
        finally:
            clean()
        self.assertEquals(20070702,xbegin)
        self.assertEquals(20070802,xend)
        self.assertEquals(23,xcount)
        self.assertEquals(71,scount)


